package mr

import (
	"encoding/json"
	"fmt"
	"log"
	"net"
	"net/http"
	"net/rpc"
	"os"
	"strings"
	"sync"
	"time"
)

type WorkerIdAndTime struct {
	WorkerId  string
	TimeStamp time.Time
}

var mapDoneFlag bool = false

var mu sync.Mutex

type Coordinator struct {
	// Your definitions here.
	workerList map[string]lWorker
	//map files
	Mfiles []string
	//reduce files
	Ffiles []string
	//key:workerId value:filename
	//workers who are running map method
	MWorker map[string]string
	//workers who are running reduce method
	RWorker map[string]string
	//reduce slice num
	reduceNum int
	//lock
}

type lWorker struct {
	// idle,busy
	status string
	// node
	isHealthy bool
	//timetamp(Healthy status)
	timeStamp time.Time
	//map task starting time
	mtimeStamp time.Time
	//reduce task starting time
	rtimeStamp time.Time
}

// Your code here -- RPC handlers for the worker to call.

// an example RPC handler.
//
// the RPC argument and reply types are defined in rpc.go.
func (c *Coordinator) RegisterWorker(args *Args, reply *Reply) error {
	mu.Lock()

	strData := args.RequestBody
	rtype := args.RequestType
	if rtype != "register" {
		fmt.Println("the register request from worker made a mistake, the type is not register")
	}
	//string to json
	var data map[string]interface{}
	err := json.Unmarshal([]byte(strData), &data)
	if err != nil {
		fmt.Println("string to json failed in file coordinator method RegisterWorker...The reason is", err)
	}
	worker := lWorker{
		status:    "idle",
		isHealthy: true,
		timeStamp: time.Now(),
	}
	// c.workerList = make(map[string]lWorker, 100)
	// c.MWorker = make(map[string]string, 100)
	// c.RWorker = make(map[string]string, 100)
	c.workerList[data["workerId"].(string)] = worker
	//add to idleWorker,waitting operate
	// fmt.Println("the coordinator situation:", c)
	reply.code = "200"
	mu.Unlock()

	return nil
}

// start a thread that listens for RPCs from worker.go
func (c *Coordinator) server() {
	rpc.Register(c)
	rpc.HandleHTTP()
	//l, e := net.Listen("tcp", ":1234")
	sockname := coordinatorSock()
	os.Remove(sockname)
	// fmt.Println("sockname in coordinator is ", sockname)
	l, e := net.Listen("unix", sockname)
	if e != nil {
		log.Fatal("listen error:", e)
	}
	go http.Serve(l, nil)
}

// main/mrcoordinator.go calls Done() periodically to find out
// if the entire job has finished.
func (c *Coordinator) Done() bool {
	var done bool = false
	mu.Lock()
	if len(c.Ffiles) > 0 || len(c.Mfiles) > 0 || len(c.MWorker) > 0 || len(c.RWorker) > 0 {
		done = false
		// fmt.Println("c.Files: ", c.Ffiles)
		// fmt.Println("c.Miles: ", c.Mfiles)
		// fmt.Println("c.Mworker: ", c.MWorker)
		// fmt.Println("c.Rworker: ", c.RWorker)
	} else {
		// fmt.Println("Bye Bye ~~~~~~~~~~~")
		done = true
	}
	mu.Unlock()
	return done

}

// allocate Map task to Worker
func (c *Coordinator) ProvideMapTask(args *Args, reply *Reply) error {
	mu.Lock()

	workerId := args.RequestBody
	var flag bool
	for index, value := range c.Mfiles {
		flag = false
		for _, workerDoingValue := range c.MWorker {
			if workerDoingValue == value {
				flag = true
				break
			}
		}
		if flag {
			continue
		}
		reply.ResponseBody = value
		c.Mfiles = append(c.Mfiles[:index], c.Mfiles[index+1:]...)
		break
	}

	if reply.ResponseBody != "" {
		worker := c.workerList[workerId]
		worker.status = "busy"
		worker.isHealthy = true
		worker.mtimeStamp = time.Now()
		c.MWorker[workerId] = reply.ResponseBody
		c.workerList[workerId] = worker
		// fmt.Println("+++++", c.MWorker, "+++++++")
	}
	// fmt.Println(">>>>>>>>>>>>>>>>>>>>>>>ProvideMapTask<<<<<<<<<<<<<<<<<<<<<<<<<")
	// fmt.Println("offered Map Task:", reply.ResponseBody)
	// fmt.Println("Provide Map Task mapfiles:", c.Mfiles)
	// fmt.Println("Provide Map Task Now the MapWorkers are ", c.MWorker)
	// fmt.Println("<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<")
	mu.Unlock()

	return nil
}

// allocate Reduce task to Worker
func (c *Coordinator) ProvideReduceTask(args *Args, reply *Reply) error {
	mu.Lock()
	workerId := args.RequestBody
	var flag bool
	for index, value := range c.Ffiles {
		flag = false
		for _, workerDoingValue := range c.RWorker {
			if workerDoingValue == value {
				flag = true
				break
			}
		}
		if flag {
			continue
		}
		reply.ResponseBody = value
		c.Ffiles = append(c.Ffiles[:index], c.Ffiles[index+1:]...)
		break
	}
	if reply.ResponseBody != "" {
		worker := c.workerList[workerId]
		worker.rtimeStamp = time.Now()
		worker.status = "busy"
		worker.isHealthy = true
		c.workerList[workerId] = worker
		c.RWorker[workerId] = reply.ResponseBody
	}
	// fmt.Println(">>>>>>>>>>>>>>>>>>>>>>>ProvideReduceTask<<<<<<<<<<<<<<<<<<<<<<<<<")
	// fmt.Println("offered Reduce Task:", reply.ResponseBody)
	// fmt.Println("Reucefiles:", c.Ffiles)
	// fmt.Println("ReuceWorkers are ", c.RWorker)
	mu.Unlock()

	return nil
}

// check if all done
// func (c *Coordinator) Finish(args *Args, reply *Reply) error {
// 	mu.Lock()
// 	if len(c.Ffiles) > 0 || len(c.Mfiles) > 0 || len(c.MWorker) > 0 || len(c.RWorker) > 0 {
// 		reply.ResponseBody = "false"
// 	} else {
// 		reply.ResponseBody = "true"
// 	}
// 	mu.Unlock()
// 	return nil
// }

// Handle Map FinSign
func (c *Coordinator) HandleMapFinSign(args *Args, reply *Reply) error {
	mu.Lock()
	var data map[string]interface{}
	err := json.Unmarshal([]byte(args.RequestBody), &data)
	if err != nil {
		panic(err)
	}
	workerId := data["workerId"].(string)
	filename := data["fileName"].(string)
	// fmt.Println(">>>>>>>>>>>>>>>>>>>>>>HandleMapFinSign>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")
	// fmt.Println("receivec workerId : ", workerId)
	// fmt.Println("received fileName :", filename)
	// fmt.Println("now the co Mfile :", c.Mfiles)
	// fmt.Println("now the co MWorker :", c.MWorker)
	for _, value := range c.MWorker {
		if value == filename {
			worker := c.workerList[workerId]
			worker.isHealthy = true
			worker.status = "idle"
			c.workerList[workerId] = worker
			delete(c.MWorker, workerId)
			reply.ResponseBody = "200"
			break
		}
	}
	// fmt.Println("after for the co Mfile :", c.Mfiles)
	// fmt.Println("after for co MWorker :", c.MWorker)
	// fmt.Println("<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<")
	mu.Unlock()

	return nil
}

func (c *Coordinator) HandleReduceFinSign(args *Args, reply *Reply) error {

	mu.Lock()
	var data map[string]interface{}
	err := json.Unmarshal([]byte(args.RequestBody), &data)
	if err != nil {
		log.Println("Failed to transfer data from string to json in heartbeat method")
	}
	workerId := data["workerId"].(string)
	filename := data["fileName"].(string)
	for _, value := range c.RWorker {
		if value == filename {
			worker := c.workerList[workerId]
			worker.isHealthy = true
			worker.status = "idle"
			c.workerList[workerId] = worker
			delete(c.RWorker, workerId)
			reply.ResponseBody = "200"
			break
		}
	}
	mu.Unlock()

	return nil
}

// check if map finished ,if finished,allocate reduce file names
func (c *Coordinator) MapFinished(args *Args, reply *Reply) error {
	mu.Lock()
	if len(c.Mfiles) > 0 || len(c.MWorker) > 0 {
		reply.ResponseBody = "false"
	} else {
		if !mapDoneFlag {
			dd, err := os.Open(".")
			if err != nil {
				panic(err)
			}
			names, err := dd.Readdirnames(1000000)
			if err != nil {
				panic(err)
			}
			for _, name := range names {
				if strings.HasPrefix(name, "mr-tmp") {
					c.Ffiles = append(c.Ffiles, name)
				}
			}
			// fmt.Println(c)
			reply.ResponseBody = "true"
			dd.Close()
			mapDoneFlag = true
			// fmt.Println("Reduce files:", c.Ffiles)
		}
		reply.ResponseBody = "true"
	}
	mu.Unlock()
	return nil
}

func (c *Coordinator) ReduceFinished(args *Args, reply *Reply) error {
	mu.Lock()
	if len(c.Ffiles) > 0 || len(c.RWorker) > 0 {
		reply.ResponseBody = "false"
	} else {
		reply.ResponseBody = "true"
	}
	mu.Unlock()

	return nil
}

// receive heart beat from worker
func (c *Coordinator) HeartBeat(args *Args, reply *Reply) error {
	mu.Lock()
	var wt WorkerIdAndTime
	err := json.Unmarshal([]byte(args.RequestBody), &wt)
	if err != nil {
		log.Println("Failed to transfer data from string to json in heartbeat method")
	}
	workerId := wt.WorkerId
	timestamp := wt.TimeStamp
	w := c.workerList[workerId]
	w.isHealthy = true
	w.timeStamp = timestamp

	c.workerList[workerId] = w
	reply.code = "200"
	mu.Unlock()

	return nil

}

// func (c *Coordinator) MonitorWorkers(durationTime time.Duration) {
// 	for {
// 		time.Sleep(durationTime / 2)
// 		for id, value := range c.workerList {
// 			if time.Since(value.timeStamp) > durationTime {
// 				c.setDeadStatus(id)
// 			}
// 		}
// 		time.Sleep(1 * time.Second)
// 	}
// }

func (c *Coordinator) MonitorWorkersMTask(durationTime time.Duration) {
	for {
		//key:the monitor workerId  value:the handling filename
		time.Sleep(2 * time.Second)
		mu.Lock()
		if len(c.MWorker) > 0 {
			// fmt.Println(">>>>>>>>>>>>>>>>>Monitor   Map   Task>>>>>>>>>>>>>>>>>>>>>>>>>>>")
			// fmt.Println("before the map workers are ", c.MWorker)
			//if some worker map phase acrossed the time,then allocate this file to another worker
			for key, _ := range c.MWorker {
				if time.Since(c.workerList[key].mtimeStamp) > durationTime {
					c.setDeadStatus(key)
					c.Mfiles = append(c.Mfiles, c.MWorker[key])
					delete(c.MWorker, key)
					continue
				}
				args := Args{}
				reply := Reply{}
				ret := callToWorker("Lworker.IsHealth", &args, &reply, key)
				if !ret {
					fmt.Println(key, "is dead..revelent messages are :", c.MWorker[key])
					c.Mfiles = append(c.Mfiles, c.MWorker[key])
					delete(c.MWorker, key)
				}
			}
			// fmt.Println("after the map workers are ", c.MWorker)
			// fmt.Println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")
		}
		mu.Unlock()
	}
}

func (c *Coordinator) MonitorWorkersRTask(durationTime time.Duration) {
	for {
		time.Sleep(2 * time.Second)
		mu.Lock()
		if len(c.RWorker) > 0 {
			// fmt.Println(">>>>>>>>>>>>>>>>Monitor Reduce Task>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")
			// fmt.Println("before the reduce workers are ", c.RWorker)
			for key, _ := range c.RWorker {
				if time.Since(c.workerList[key].rtimeStamp) > durationTime {
					c.setDeadStatus(key)
					c.Ffiles = append(c.Ffiles, c.RWorker[key])
					delete(c.RWorker, key)
				}
				args := Args{}
				reply := Reply{}
				ret := callToWorker("Lworker.IsHealth", &args, &reply, key)
				if !ret {
					c.Ffiles = append(c.Ffiles, c.RWorker[key])
					delete(c.RWorker, key)
				}
			}
			// fmt.Println("after the reduce workers are ", c.RWorker)
		}
		mu.Unlock()
	}
}

func (c *Coordinator) setDeadStatus(workerId string) {
	fmt.Println("<<<<<<<<<<<<  Set Dead Status<<<<<<<<<<<<<<<<")
	worker := c.workerList[workerId]
	worker.status = "dead"
	worker.isHealthy = false
	c.workerList[workerId] = worker
	args := Args{
		RequestBody: workerId,
	}
	reply := Reply{}
	callToWorker("Lworker.Exit", &args, &reply, workerId)
	// fmt.Println(workerId, " has Exited")
	// fmt.Println("<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<")
}

// create a Coordinator.
// main/mrcoordinator.go calls this function.
// nReduce is the number of reduce tasks to use.
func MakeCoordinator(files []string, nReduce int) *Coordinator {
	c := Coordinator{Mfiles: files, reduceNum: nReduce}
	c.workerList = make(map[string]lWorker, 100)
	c.MWorker = make(map[string]string, 100)
	c.RWorker = make(map[string]string, 100)
	// Your code here.
	c.server()
	// go c.MonitorWorkers(10)
	go c.MonitorWorkersMTask(10 * time.Second)
	go c.MonitorWorkersRTask(10 * time.Second)
	return &c
}

// args :{ wokerId }
func callToWorker(rpcname string, args interface{}, reply interface{}, workerId string) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	// build sockname
	sockname := workerSock() + workerId
	co, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Println(err)
		return false
	}
	defer co.Close()

	err = co.Call(rpcname, args, reply)
	if err == nil {
		return true
	} else {
		return false
	}
}
